package com.ict.dataprocess;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

import org.apache.log4j.Logger;

import com.google.protobuf.InvalidProtocolBufferException;
import com.ict.config.Config;
import com.ict.utils.CacheObject;
import com.ict.utils.ConnMQUtil;
import com.ict.utils.ConnRedisUtil;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

import team.iOceanPlus.PB.AISData.PBAISDynamic;
import team.iOceanPlus.PB.CommonEnum.PB_Enum_TargetInfo_Type;
import team.iOceanPlus.PB.Target;
import team.iOceanPlus.PB.Target.PBTargetPosition;

public class OrigMQToRedis extends Thread {
	private static Logger logger = Logger.getLogger(OrigMQToRedis.class);

	private boolean isStopped = false;

	public void run() {
		List<byte []>  lstCache=CacheObject.returnLstCache();
		if (lstCache.size()==Config.listCatchSize) {
			SyncFlagClass isFlag = new SyncFlagClass();
			List<Target.PBTargetPosition> lstRecvPBTargetPos = new ArrayList<Target.PBTargetPosition>();
			// ConnRedisUtil.jedis.
			//Map<byte[], byte[]> rt_map = ConnRedisUtil.jedis.hgetAll("recv_orig_pb".getBytes());
			// ConnRedisUtil.jedis.del("recv_orig_pb");
			/*
			 * for(int x=0;x<CacheObject.lstBytCache.size();x++) { try {
			 * lstRecvPBTargetPos.addAll(Target.PBTarget.parseFrom(CacheObject.lstBytCache.
			 * get(x)).getListTargetPosList()); CacheObject.lstBytCache } catch
			 * (InvalidProtocolBufferException e) { // TODO Auto-generated catch block
			 * e.printStackTrace(); }
			 * 
			 * }
			 */
			for (byte[] byteKey : lstCache) {
				// rt_map.get(byteKey);
				try {
					// ConnRedisUtil.jedis.hdel("recv_orig_pb".getBytes(), byteKey);
					lstRecvPBTargetPos.addAll(Target.PBTarget.parseFrom(byteKey).getListTargetPosList());
					// CacheObject.lstBytCache.remove(byteKey);
				} catch (InvalidProtocolBufferException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			/*
			 * for (byte[] byteArray : rt_map.values()) { try {
			 * lstRecvPBTargetPos.addAll(Target.PBTarget.parseFrom(byteArray).
			 * getListTargetPosList()); } catch (InvalidProtocolBufferException e) { // TODO
			 * Auto-generated catch block e.printStackTrace(); } }
			 */

			logger.info("orig--position count is" + lstRecvPBTargetPos.size());
			try {
				for (int x = 0; x < lstRecvPBTargetPos.size(); x++) {
					Target.PBTargetPosition pos = lstRecvPBTargetPos.get(x);
					if (pos.getEnumTargetInfoType()
							.getNumber() == PB_Enum_TargetInfo_Type.EV_TargetInfoType_AISStatic_VALUE) {
						lstRecvPBTargetPos.remove(x);
						x = x - 1;
					}

				}
				/** 去掉重复数据 */
				lstRecvPBTargetPos.stream().distinct().collect(Collectors.toList());
				/** 同一目标位置信息保存最新时间一条 */
				/*
				 * List<Target.PBTargetPosition> group = new
				 * ArrayList<Target.PBTargetPosition>();
				 */

				for (Target.PBTargetPosition pos : lstRecvPBTargetPos) {
					String targetIDOrig = pos.getTargetIDOrig() + "";
					String infoTypeNumber = pos.getEnumTargetInfoTypeValue() + "";
					;
					String infoSourceNumber = pos.getEnumTargetInfoSourceValue() + "";
					StringBuffer fieldKey = new StringBuffer();
					fieldKey.append(targetIDOrig).append("+").append(infoTypeNumber).append("+")
							.append(infoSourceNumber);
					/*byte [] bytFeild=fieldKey.toString().getBytes();
					if (ConnRedisUtil.jedis.hexists("mid_pos_new", fieldKey.toString())) {
						byte[] bytRedisPos = ConnRedisUtil.jedis.hget("mid_pos_new".getBytes(),
								bytFeild);
						Target.PBTargetPosition redisPos = (Target.PBTargetPosition) byteArrayToObject(bytRedisPos);
						if (pos.getLastAssocUsingDSCombinationUTCTime() > redisPos
								.getLastAssocUsingDSCombinationUTCTime()) {
							ConnRedisUtil.jedis.hset("mid_pos_new".getBytes(), fieldKey.toString().getBytes(),
									objectToByteArray(pos));
						}

					} else {*/
						ConnRedisUtil.jedis.hset("mid_pos_new".getBytes(), fieldKey.toString().getBytes(),
								objectToByteArray(pos));
					//}

				}

				/*
				 * Map<byte[] ,byte[]> newMap=
				 * ConnRedisUtil.jedis.hgetAll("mid_pos_new".getBytes()); Map<byte[] ,byte[]>
				 * oldMap= ConnRedisUtil.jedis.hgetAll("mid_pos_old".getBytes());
				 */
				/** 目标点没有变化的移除 */
				/*
				 * for(byte[] bytPos :newMap.values()) { group.add((Target.PBTargetPosition)
				 * byteArrayToObject(bytPos)); }
				 */
				/* ConnRedisUtil.jedis.rename("mid_pos_new", "send_orig_pb"); */
				/*
				 * logger.info("orig--position filter count is"+group.size());
				 * 
				 * for( int x=0;x<group.size() ;x++) {
				 * ConnRedisUtil.jedis.hset("send_orig_pb".getBytes(), ("field"+x).getBytes(),
				 * objectToByteArray(group.get(x))); }
				 */
				/*
				 * Target.PBTarget.Builder targetBuilder = Target.PBTarget.newBuilder();
				 * targetBuilder.setEnumSenderSoftwareValue(6);
				 * targetBuilder.addAllListTargetPos(group); Target.PBTarget pbTarget =
				 * targetBuilder.build(); ConnRedisUtil.jedis.hset("send_orig_pb".getBytes(),
				 * ("field").getBytes(), pbTarget.toByteArray());
				 */
				/** 融合模式 */
				/*
				 * String[] arrRoutingKey2 = { "OnLine.FusedData.TargetTracks" }; ConnMQUtil
				 * connMQUtil2 = new ConnMQUtil(); QueueingConsumer consumer2;
				 * 
				 * consumer = connMQUtil.recv("topic", "xgs", arrRoutingKey2, "xgs_tap_by_tlc");
				 * QueueingConsumer.Delivery delivery2 = consumer.nextDelivery();
				 * Target.PBTarget info2 = Target.PBTarget.parseFrom(delivery2.getBody());
				 *//** 去掉AIS静态数据 */
				/*
				 * for (int x = 0; x < info2.getListTargetPosList().size(); x++) {
				 * Target.PBTargetPosition pos= info2.getListTargetPosList().get(x);
				 * if(pos.getEnumTargetInfoType().getNumber()==3) {
				 * info2.getListTargetPosList().remove(x);
				 * 
				 * 
				 * } }
				 *//** 去掉重复数据 */
				/*
				 * info2.getListTargetPosList().stream().distinct().collect(Collectors.toList())
				 * ;
				 *//** 同一目标位置信息保存最新时间一条 */
				/*
				 * List<Target.PBTargetPosition> group2 = (List<PBTargetPosition>)
				 * info2.getListTargetPosList().stream().limit(1).collect(Collectors.groupingBy(
				 * t -> t.getTargetID()));
				 *//** 目标点没有变化的移除 *//*
									 * for (int x = 0; x < group.size(); x++) { Target.PBTargetPosition pos=
									 * info2.getListTargetPosList().get(x); Boolean
									 * isContainsPos=CacheObject.lstOrigTargetCache.get(0).getListTargetPosList().
									 * contains(pos); if(isContainsPos) { group2.remove(group.get(x)); } }
									 * CacheObject.lstTargetCache.clear(); CacheObject.lstTargetCache.add(info2);
									 * System.out.println("recving redis");
									 * ConnMQUtil.recvChannel.basicAck(delivery.getEnvelope().getDeliveryTag(),
									 * false);
									 */

			} catch (ShutdownSignalException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			} catch (ConsumerCancelledException e1) {
				// TODO Auto-generated catch block
				e1.printStackTrace();
			}

			System.out.println("second over");
			OrigRedisToMQ third = new OrigRedisToMQ();
			third.run();
			lstCache.clear();
		}
	}

	public void stopThread() {
		isStopped = true;
	}

	public void startThread() {
		isStopped = false;
	}

	public static byte[] objectToByteArray(Object obj) {
		byte[] bytes = null;
		ByteArrayOutputStream byteArrayOutputStream = null;
		ObjectOutputStream objectOutputStream = null;
		try {
			byteArrayOutputStream = new ByteArrayOutputStream();
			objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);
			objectOutputStream.writeObject(obj);
			objectOutputStream.flush();
			bytes = byteArrayOutputStream.toByteArray();

		} catch (IOException e) {
		} finally {
			if (objectOutputStream != null) {
				try {
					objectOutputStream.close();
				} catch (IOException e) {
				}
			}
			if (byteArrayOutputStream != null) {
				try {
					byteArrayOutputStream.close();
				} catch (IOException e) {
				}
			}

		}
		return bytes;
	}

	public static Object byteArrayToObject(byte[] bytes) {
		Object obj = null;
		ByteArrayInputStream byteArrayInputStream = null;
		ObjectInputStream objectInputStream = null;
		try {
			byteArrayInputStream = new ByteArrayInputStream(bytes);
			objectInputStream = new ObjectInputStream(byteArrayInputStream);
			obj = objectInputStream.readObject();
		} catch (Exception e) {
		} finally {
			if (byteArrayInputStream != null) {
				try {
					byteArrayInputStream.close();
				} catch (IOException e) {
				}
			}
			if (objectInputStream != null) {
				try {
					objectInputStream.close();
				} catch (IOException e) {
				}
			}
		}
		return obj;
	}

}
